#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Excel文件处理器

该模块提供了Excel文件的读写和处理功能。

创建时间: 2023-07-12
作者: Python Tools Team
"""

import pandas as pd
import numpy as np
from typing import Optional, Union, Dict, Any, List, Tuple, Callable, Set
from pathlib import Path
import io
import os

from modules.logger import LogManager
from modules.exception import exception_handler, FileException
from modules.fileio.file_manager import FileManager


class ExcelFileHandler:
    """
    Excel文件处理器
    
    该类提供了Excel文件的读写和处理功能，支持多种格式和选项。
    """
    
    def __init__(self, file_manager: Optional[FileManager] = None, logger=None):
        """
        初始化Excel文件处理器
        
        Args:
            file_manager: 文件管理器实例，如果为None则创建新实例
            logger: 日志记录器，如果为None则使用默认日志记录器
        """
        self.file_manager = file_manager or FileManager()
        self.logger = logger or LogManager.get_both_logger(name="excel_handler")
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def read(self, file_path: Union[str, Path], sheet_name=0, **kwargs) -> pd.DataFrame:
        """
        读取Excel文件为DataFrame
        
        Args:
            file_path: Excel文件路径
            sheet_name: 工作表名称或索引，默认为0（第一个工作表）
            **kwargs: 传递给pandas.read_excel的其他参数
        
        Returns:
            pd.DataFrame: 读取的DataFrame
        
        Raises:
            FileException: 文件不存在或读取失败时抛出
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx', sheet_name='Sheet1')
            >>> print(df.head())
        """
        file_path = Path(file_path)
        if not file_path.exists():
            raise FileException(f"文件不存在: {file_path}")
        
        try:
            df = pd.read_excel(file_path, sheet_name=sheet_name, **kwargs)
        except Exception as e:
            raise FileException(f"读取Excel文件失败: {e}")
        
        self.logger.debug(f"已读取Excel文件: {file_path}, 工作表: {sheet_name}")
        return df
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def read_all_sheets(self, file_path: Union[str, Path], **kwargs) -> Dict[str, pd.DataFrame]:
        """
        读取Excel文件中的所有工作表
        
        Args:
            file_path: Excel文件路径
            **kwargs: 传递给pandas.read_excel的其他参数
        
        Returns:
            Dict[str, pd.DataFrame]: 工作表名称到DataFrame的映射
        
        Raises:
            FileException: 文件不存在或读取失败时抛出
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> sheets = excel_handler.read_all_sheets('data.xlsx')
            >>> for name, df in sheets.items():
            >>>     print(f"Sheet: {name}")
            >>>     print(df.head())
        """
        file_path = Path(file_path)
        if not file_path.exists():
            raise FileException(f"文件不存在: {file_path}")
        
        try:
            sheets = pd.read_excel(file_path, sheet_name=None, **kwargs)
        except Exception as e:
            raise FileException(f"读取Excel文件失败: {e}")
        
        self.logger.debug(f"已读取Excel文件的所有工作表: {file_path}")
        return sheets
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def write(self, file_path: Union[str, Path], df: pd.DataFrame, sheet_name: str = 'Sheet1', 
             create_dirs: bool = True, **kwargs) -> bool:
        """
        将DataFrame写入Excel文件
        
        Args:
            file_path: Excel文件路径
            df: 要写入的DataFrame
            sheet_name: 工作表名称，默认为'Sheet1'
            create_dirs: 如果父目录不存在，是否创建，默认为True
            **kwargs: 传递给pandas.DataFrame.to_excel的其他参数
        
        Returns:
            bool: 写入成功返回True
        
        Raises:
            FileException: 写入失败时抛出
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})
            >>> excel_handler.write('data.xlsx', df, sheet_name='Data')
        """
        file_path = Path(file_path)
        
        # 确保父目录存在
        if create_dirs:
            file_path.parent.mkdir(parents=True, exist_ok=True)
        
        try:
            df.to_excel(file_path, sheet_name=sheet_name, **kwargs)
        except Exception as e:
            raise FileException(f"写入Excel文件失败: {e}")
        
        self.logger.debug(f"已写入Excel文件: {file_path}, 工作表: {sheet_name}")
        return True
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def write_multiple_sheets(self, file_path: Union[str, Path], 
                            sheets: Dict[str, pd.DataFrame], 
                            create_dirs: bool = True, **kwargs) -> bool:
        """
        将多个DataFrame写入Excel文件的不同工作表
        
        Args:
            file_path: Excel文件路径
            sheets: 工作表名称到DataFrame的映射
            create_dirs: 如果父目录不存在，是否创建，默认为True
            **kwargs: 传递给pandas.ExcelWriter的其他参数
        
        Returns:
            bool: 写入成功返回True
        
        Raises:
            FileException: 写入失败时抛出
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df1 = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})
            >>> df2 = pd.DataFrame({'X': [10, 20, 30], 'Y': [40, 50, 60]})
            >>> excel_handler.write_multiple_sheets('data.xlsx', {'Data1': df1, 'Data2': df2})
        """
        file_path = Path(file_path)
        
        # 确保父目录存在
        if create_dirs:
            file_path.parent.mkdir(parents=True, exist_ok=True)
        
        try:
            with pd.ExcelWriter(file_path, **kwargs) as writer:
                for sheet_name, df in sheets.items():
                    df.to_excel(writer, sheet_name=sheet_name)
        except Exception as e:
            raise FileException(f"写入Excel文件失败: {e}")
        
        self.logger.debug(f"已写入Excel文件的多个工作表: {file_path}")
        return True
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def append_sheet(self, file_path: Union[str, Path], df: pd.DataFrame, 
                    sheet_name: str, **kwargs) -> bool:
        """
        向现有Excel文件添加新工作表
        
        Args:
            file_path: Excel文件路径
            df: 要添加的DataFrame
            sheet_name: 新工作表名称
            **kwargs: 传递给pandas.DataFrame.to_excel的其他参数
        
        Returns:
            bool: 添加成功返回True
        
        Raises:
            FileException: 文件不存在或添加失败时抛出
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = pd.DataFrame({'C': [7, 8, 9], 'D': [10, 11, 12]})
            >>> excel_handler.append_sheet('data.xlsx', df, 'NewData')
        """
        file_path = Path(file_path)
        if not file_path.exists():
            raise FileException(f"文件不存在: {file_path}")
        
        try:
            # 读取现有工作表
            existing_sheets = pd.read_excel(file_path, sheet_name=None)
            
            # 检查工作表名称是否已存在
            if sheet_name in existing_sheets:
                self.logger.warning(f"工作表已存在，将被覆盖: {sheet_name}")
            
            # 添加新工作表
            with pd.ExcelWriter(file_path, mode='a', if_sheet_exists='replace') as writer:
                df.to_excel(writer, sheet_name=sheet_name, **kwargs)
        except Exception as e:
            raise FileException(f"添加工作表失败: {e}")
        
        self.logger.debug(f"已向Excel文件添加工作表: {file_path}, 工作表: {sheet_name}")
        return True
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def get_sheet_names(self, file_path: Union[str, Path]) -> List[str]:
        """
        获取Excel文件中的所有工作表名称
        
        Args:
            file_path: Excel文件路径
        
        Returns:
            List[str]: 工作表名称列表
        
        Raises:
            FileException: 文件不存在或读取失败时抛出
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> sheet_names = excel_handler.get_sheet_names('data.xlsx')
            >>> print(sheet_names)
        """
        file_path = Path(file_path)
        if not file_path.exists():
            raise FileException(f"文件不存在: {file_path}")
        
        try:
            xl = pd.ExcelFile(file_path)
            sheet_names = xl.sheet_names
        except Exception as e:
            raise FileException(f"获取工作表名称失败: {e}")
        
        return sheet_names
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def filter_data(self, df: pd.DataFrame, conditions: Dict[str, Any]) -> pd.DataFrame:
        """
        根据条件筛选DataFrame
        
        Args:
            df: 要筛选的DataFrame
            conditions: 列名到筛选条件的映射
        
        Returns:
            pd.DataFrame: 筛选后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> filtered = excel_handler.filter_data(df, {'Age': lambda x: x > 30, 'City': 'Beijing'})
            >>> print(filtered)
        """
        result = df.copy()
        
        for column, condition in conditions.items():
            if column not in result.columns:
                self.logger.warning(f"列不存在，跳过筛选条件: {column}")
                continue
            
            if callable(condition):
                # 使用函数条件
                result = result[condition(result[column])]
            else:
                # 使用相等条件
                result = result[result[column] == condition]
        
        return result
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def sort_data(self, df: pd.DataFrame, by: Union[str, List[str]], 
                ascending: Union[bool, List[bool]] = True) -> pd.DataFrame:
        """
        对DataFrame进行排序
        
        Args:
            df: 要排序的DataFrame
            by: 排序依据的列名或列名列表
            ascending: 是否升序排序，可以是单个布尔值或布尔值列表
        
        Returns:
            pd.DataFrame: 排序后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> sorted_df = excel_handler.sort_data(df, ['Age', 'Salary'], [False, True])
            >>> print(sorted_df)
        """
        return df.sort_values(by=by, ascending=ascending)
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def group_data(self, df: pd.DataFrame, by: Union[str, List[str]], 
                 agg_funcs: Dict[str, Union[str, List[str]]] = None) -> pd.DataFrame:
        """
        对DataFrame进行分组聚合
        
        Args:
            df: 要分组的DataFrame
            by: 分组依据的列名或列名列表
            agg_funcs: 列名到聚合函数的映射，如果为None则只分组不聚合
        
        Returns:
            pd.DataFrame: 分组聚合后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> grouped = excel_handler.group_data(df, 'Department', {'Salary': ['mean', 'sum'], 'Age': 'mean'})
            >>> print(grouped)
        """
        grouped = df.groupby(by)
        
        if agg_funcs is None:
            return grouped.size().reset_index(name='count')
        
        return grouped.agg(agg_funcs).reset_index()
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def pivot_data(self, df: pd.DataFrame, index: Union[str, List[str]], 
                 columns: Union[str, List[str]], values: Union[str, List[str]], 
                 aggfunc: Union[str, Callable] = 'mean') -> pd.DataFrame:
        """
        创建数据透视表
        
        Args:
            df: 源DataFrame
            index: 行索引的列名或列名列表
            columns: 列索引的列名或列名列表
            values: 值的列名或列名列表
            aggfunc: 聚合函数，默认为'mean'
        
        Returns:
            pd.DataFrame: 数据透视表
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> pivot = excel_handler.pivot_data(df, 'Department', 'Year', 'Salary', 'sum')
            >>> print(pivot)
        """
        return pd.pivot_table(df, index=index, columns=columns, values=values, aggfunc=aggfunc)
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def merge_dataframes(self, left: pd.DataFrame, right: pd.DataFrame, 
                       how: str = 'inner', on: Union[str, List[str]] = None, 
                       left_on: Union[str, List[str]] = None, 
                       right_on: Union[str, List[str]] = None) -> pd.DataFrame:
        """
        合并两个DataFrame
        
        Args:
            left: 左侧DataFrame
            right: 右侧DataFrame
            how: 合并方式，可选'inner', 'outer', 'left', 'right'，默认为'inner'
            on: 合并依据的列名或列名列表
            left_on: 左侧DataFrame的合并列
            right_on: 右侧DataFrame的合并列
        
        Returns:
            pd.DataFrame: 合并后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df1 = excel_handler.read('employees.xlsx')
            >>> df2 = excel_handler.read('departments.xlsx')
            >>> merged = excel_handler.merge_dataframes(df1, df2, how='left', left_on='DeptID', right_on='ID')
            >>> print(merged)
        """
        return pd.merge(left, right, how=how, on=on, left_on=left_on, right_on=right_on)
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def concat_dataframes(self, dfs: List[pd.DataFrame], axis: int = 0, 
                        ignore_index: bool = False) -> pd.DataFrame:
        """
        连接多个DataFrame
        
        Args:
            dfs: DataFrame列表
            axis: 连接轴，0为行连接，1为列连接，默认为0
            ignore_index: 是否忽略原索引，默认为False
        
        Returns:
            pd.DataFrame: 连接后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df1 = excel_handler.read('data2020.xlsx')
            >>> df2 = excel_handler.read('data2021.xlsx')
            >>> df3 = excel_handler.read('data2022.xlsx')
            >>> combined = excel_handler.concat_dataframes([df1, df2, df3], ignore_index=True)
            >>> print(combined)
        """
        return pd.concat(dfs, axis=axis, ignore_index=ignore_index)
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def apply_formula(self, df: pd.DataFrame, column: str, formula: Callable, 
                    new_column: Optional[str] = None) -> pd.DataFrame:
        """
        对DataFrame应用公式
        
        Args:
            df: 要处理的DataFrame
            column: 要应用公式的列名
            formula: 公式函数
            new_column: 结果列名，如果为None则覆盖原列
        
        Returns:
            pd.DataFrame: 处理后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 将温度从摄氏度转换为华氏度
            >>> df = excel_handler.apply_formula(df, 'Temperature', lambda x: x * 9/5 + 32, 'Temperature_F')
            >>> print(df)
        """
        result = df.copy()
        
        if column not in result.columns:
            raise ValueError(f"列不存在: {column}")
        
        if new_column is None:
            result[column] = result[column].apply(formula)
        else:
            result[new_column] = result[column].apply(formula)
        
        return result
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def fill_missing_values(self, df: pd.DataFrame, columns: Union[str, List[str]] = None, 
                          method: str = 'ffill', value: Any = None) -> pd.DataFrame:
        """
        填充缺失值
        
        Args:
            df: 要处理的DataFrame
            columns: 要填充的列名或列名列表，如果为None则填充所有列
            method: 填充方法，可选'ffill', 'bfill', 'value'，默认为'ffill'
            value: 当method为'value'时使用的填充值
        
        Returns:
            pd.DataFrame: 填充后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 使用前向填充
            >>> df1 = excel_handler.fill_missing_values(df, ['A', 'B'], method='ffill')
            >>> # 使用指定值填充
            >>> df2 = excel_handler.fill_missing_values(df, 'C', method='value', value=0)
            >>> print(df2)
        """
        result = df.copy()
        
        if columns is None:
            columns = result.columns
        elif isinstance(columns, str):
            columns = [columns]
        
        for column in columns:
            if column not in result.columns:
                self.logger.warning(f"列不存在，跳过填充: {column}")
                continue
            
            if method == 'ffill':
                result[column] = result[column].fillna(method='ffill')
            elif method == 'bfill':
                result[column] = result[column].fillna(method='bfill')
            elif method == 'value':
                result[column] = result[column].fillna(value)
            else:
                raise ValueError(f"不支持的填充方法: {method}")
        
        return result
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def remove_duplicates(self, df: pd.DataFrame, subset: Union[str, List[str]] = None, 
                        keep: str = 'first') -> pd.DataFrame:
        """
        移除重复行
        
        Args:
            df: 要处理的DataFrame
            subset: 用于判断重复的列名或列名列表，如果为None则使用所有列
            keep: 保留哪个重复项，可选'first', 'last', False，默认为'first'
        
        Returns:
            pd.DataFrame: 处理后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 根据ID列移除重复行，保留第一次出现的
            >>> df = excel_handler.remove_duplicates(df, 'ID')
            >>> print(df)
        """
        return df.drop_duplicates(subset=subset, keep=keep)
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def convert_column_type(self, df: pd.DataFrame, column: str, 
                          dtype: Union[str, np.dtype, type]) -> pd.DataFrame:
        """
        转换列的数据类型
        
        Args:
            df: 要处理的DataFrame
            column: 要转换的列名
            dtype: 目标数据类型
        
        Returns:
            pd.DataFrame: 处理后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 将字符串列转换为整数
            >>> df = excel_handler.convert_column_type(df, 'ID', int)
            >>> print(df.dtypes)
        """
        result = df.copy()
        
        if column not in result.columns:
            raise ValueError(f"列不存在: {column}")
        
        try:
            result[column] = result[column].astype(dtype)
        except Exception as e:
            raise ValueError(f"类型转换失败: {e}")
        
        return result
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def rename_columns(self, df: pd.DataFrame, columns: Dict[str, str]) -> pd.DataFrame:
        """
        重命名列
        
        Args:
            df: 要处理的DataFrame
            columns: 旧列名到新列名的映射
        
        Returns:
            pd.DataFrame: 处理后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> df = excel_handler.rename_columns(df, {'Name': 'FullName', 'Dept': 'Department'})
            >>> print(df.columns)
        """
        return df.rename(columns=columns)
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def add_column(self, df: pd.DataFrame, column: str, value: Any) -> pd.DataFrame:
        """
        添加新列
        
        Args:
            df: 要处理的DataFrame
            column: 新列名
            value: 列值，可以是标量、列表、Series或函数
        
        Returns:
            pd.DataFrame: 处理后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 添加常量列
            >>> df = excel_handler.add_column(df, 'Status', 'Active')
            >>> # 添加计算列
            >>> df = excel_handler.add_column(df, 'Total', df['Price'] * df['Quantity'])
            >>> print(df)
        """
        result = df.copy()
        result[column] = value
        return result
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def drop_columns(self, df: pd.DataFrame, columns: Union[str, List[str]]) -> pd.DataFrame:
        """
        删除列
        
        Args:
            df: 要处理的DataFrame
            columns: 要删除的列名或列名列表
        
        Returns:
            pd.DataFrame: 处理后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> df = excel_handler.drop_columns(df, ['Temp1', 'Temp2'])
            >>> print(df.columns)
        """
        return df.drop(columns=columns)
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def filter_rows(self, df: pd.DataFrame, condition: Callable[[pd.Series], bool]) -> pd.DataFrame:
        """
        根据条件筛选行
        
        Args:
            df: 要处理的DataFrame
            condition: 行筛选条件函数
        
        Returns:
            pd.DataFrame: 处理后的DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 筛选年龄大于30的行
            >>> df = excel_handler.filter_rows(df, lambda row: row['Age'] > 30)
            >>> print(df)
        """
        return df[df.apply(condition, axis=1)]
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def to_csv(self, df: pd.DataFrame, file_path: Union[str, Path], 
              encoding: str = 'utf-8', **kwargs) -> bool:
        """
        将DataFrame导出为CSV文件
        
        Args:
            df: 要导出的DataFrame
            file_path: CSV文件路径
            encoding: 文件编码，默认为utf-8
            **kwargs: 传递给pandas.DataFrame.to_csv的其他参数
        
        Returns:
            bool: 导出成功返回True
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> excel_handler.to_csv(df, 'data.csv', index=False)
        """
        file_path = Path(file_path)
        
        try:
            df.to_csv(file_path, encoding=encoding, **kwargs)
        except Exception as e:
            raise FileException(f"导出CSV文件失败: {e}")
        
        self.logger.debug(f"已导出DataFrame到CSV文件: {file_path}")
        return True
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def to_json(self, df: pd.DataFrame, file_path: Union[str, Path], 
               orient: str = 'records', **kwargs) -> bool:
        """
        将DataFrame导出为JSON文件
        
        Args:
            df: 要导出的DataFrame
            file_path: JSON文件路径
            orient: JSON格式，默认为'records'
            **kwargs: 传递给pandas.DataFrame.to_json的其他参数
        
        Returns:
            bool: 导出成功返回True
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> excel_handler.to_json(df, 'data.json')
        """
        file_path = Path(file_path)
        
        try:
            df.to_json(file_path, orient=orient, **kwargs)
        except Exception as e:
            raise FileException(f"导出JSON文件失败: {e}")
        
        self.logger.debug(f"已导出DataFrame到JSON文件: {file_path}")
        return True
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def compare_sheets(self, file_path1: Union[str, Path], file_path2: Union[str, Path], 
                     sheet_name: Union[str, int] = 0) -> pd.DataFrame:
        """
        比较两个Excel文件中的工作表
        
        Args:
            file_path1: 第一个Excel文件路径
            file_path2: 第二个Excel文件路径
            sheet_name: 要比较的工作表名称或索引，默认为0
        
        Returns:
            pd.DataFrame: 差异DataFrame
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> diff = excel_handler.compare_sheets('data_v1.xlsx', 'data_v2.xlsx', 'Sheet1')
            >>> print(diff)
        """
        df1 = self.read(file_path1, sheet_name=sheet_name)
        df2 = self.read(file_path2, sheet_name=sheet_name)
        
        # 确保两个DataFrame有相同的列
        common_columns = set(df1.columns).intersection(set(df2.columns))
        if not common_columns:
            raise ValueError("两个工作表没有共同的列")
        
        # 比较共同列
        df1_subset = df1[list(common_columns)]
        df2_subset = df2[list(common_columns)]
        
        # 找出差异
        df_diff = pd.concat([df1_subset, df2_subset]).drop_duplicates(keep=False)
        
        return df_diff
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def validate_data(self, df: pd.DataFrame, rules: Dict[str, Callable[[pd.Series], pd.Series]]) -> Dict[str, pd.DataFrame]:
        """
        验证DataFrame数据
        
        Args:
            df: 要验证的DataFrame
            rules: 列名到验证规则函数的映射
        
        Returns:
            Dict[str, pd.DataFrame]: 列名到验证失败行的映射
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 定义验证规则
            >>> rules = {
            >>>     'Age': lambda x: x > 0,  # 年龄必须大于0
            >>>     'Email': lambda x: x.str.contains('@')  # 邮箱必须包含@
            >>> }
            >>> invalid = excel_handler.validate_data(df, rules)
            >>> for column, invalid_rows in invalid.items():
            >>>     print(f"Invalid {column}:")
            >>>     print(invalid_rows)
        """
        invalid_data = {}
        
        for column, rule in rules.items():
            if column not in df.columns:
                self.logger.warning(f"列不存在，跳过验证: {column}")
                continue
            
            # 应用验证规则
            valid_mask = rule(df[column])
            invalid_rows = df[~valid_mask]
            
            if not invalid_rows.empty:
                invalid_data[column] = invalid_rows
        
        return invalid_data
    
    @exception_handler(exception_type=Exception, reraise_as=FileException)
    def apply_style(self, df: pd.DataFrame, file_path: Union[str, Path], 
                   style_function: Callable[[pd.DataFrame], pd.io.formats.style.Styler], 
                   sheet_name: str = 'Sheet1', **kwargs) -> bool:
        """
        应用样式并保存Excel文件
        
        Args:
            df: 要处理的DataFrame
            file_path: Excel文件路径
            style_function: 样式函数
            sheet_name: 工作表名称，默认为'Sheet1'
            **kwargs: 传递给pandas.DataFrame.to_excel的其他参数
        
        Returns:
            bool: 保存成功返回True
        
        Examples:
            >>> excel_handler = ExcelFileHandler()
            >>> df = excel_handler.read('data.xlsx')
            >>> # 定义样式函数
            >>> def highlight_max(df):
            >>>     return df.style.highlight_max(color='yellow')
            >>> excel_handler.apply_style(df, 'styled_data.xlsx', highlight_max)
        """
        file_path = Path(file_path)
        
        try:
            # 应用样式
            styled_df = style_function(df)
            
            # 保存到Excel
            styled_df.to_excel(file_path, sheet_name=sheet_name, **kwargs)
        except Exception as e:
            raise FileException(f"应用样式并保存Excel文件失败: {e}")
        
        self.logger.debug(f"已应用样式并保存Excel文件: {file_path}")
        return True